package gupao.concurrency.demo.risk.exct;

import gupao.concurrency.demo.risk.exct.processor.ExctProcessor;
import gupao.concurrency.demo.risk.model.ProcessDataModel;
import gupao.concurrency.demo.risk.model.RiskJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

@Slf4j
@Service
public class ExctService {

    private static final int RUNNING = 1;
    private static final int STANDBY = 0;

    private static final Function<RiskJob, List<Double>> edsFunc = ExctProcessor::exctEdsEnt;
    private static final Function<RiskJob, List<Double>> edcFunc = ExctProcessor::exctEdcEnt;
    private volatile int status = STANDBY;

    private final AtomicInteger edcThreadIndex = new AtomicInteger(0);
    private final AtomicInteger edsThreadIndex = new AtomicInteger(0);

    @Autowired
    private ProcessDataModel processDataModel;

    public void exct(RiskJob job) {
        if (status != STANDBY) {
            log.error("cannot do exct because current status is {}", status);
            return;
        }
        status = RUNNING;
        try {
            ExctResult result = new ExctResult();
            result.setJob(job);
            Thread t1 = new Thread(() -> {
                log.info("exct eds event data");
                List<Double> edsList = edsFunc.apply(job);
                result.setEdsDataList(edsList);
            }, "eds-thread-" + edsThreadIndex.incrementAndGet());
            t1.start();

            Thread t2 = new Thread(() -> {
                log.info("exct edc event data");
                List<Double> edcList = edcFunc.apply(job);
                result.setEdcDataList(edcList);
            }, "edc-thread-" + edcThreadIndex.incrementAndGet());
            t2.start();
            t1.join();
            t2.join();
            log.info("job {} exct finished", result.getJob().getId());
            processDataModel.getExctResultsQueue().put(result);
        } catch (InterruptedException e) {
            log.warn("exct thread interrupted, cancel risk job: {}", job.getId());
            //省略：记录audit table 便于后续重试
        }finally {
            status = STANDBY;
        }
    }
}
